package org.shj.spark.core;

import java.util.Arrays;
import java.util.Iterator;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

public class WordCount {

	public static void main(String[] args) {
		//setMaster() 设置运行模式
		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("wc");
		JavaSparkContext ctx = new JavaSparkContext(conf);
		JavaRDD<String> text = ctx.textFile("D:\\workspace\\scala\\sparkscala\\sparkscala\\file\\wordcnt.txt");//每一行的数据是 String
		//ctx.sc().
		//new FlatMapFunction<原始的类型, 期望每行转化后的类型>
		//把每一行转换成一个一个word
		JavaRDD<String> words = text.flatMap(new FlatMapFunction<String,String>(){
			private static final long serialVersionUID = -7745993570469553423L;

			public Iterator<String> call(String line) throws Exception {
				return Arrays.asList(line.split("\\s+")).iterator();
			}
		});
		
		//每一个word 计一次
		JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
			private static final long serialVersionUID = 3205165890177307621L;

			public Tuple2<String, Integer> call(String word) throws Exception {
				return new Tuple2<String, Integer>(word, 1);
			}
			
		});
		
		//Function2 前两个为输入参数，最后一个是输出参数
		JavaPairRDD<String, Integer> results = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = -5506869299961654034L;

			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1 + v2;
			}
		});
		
		//把键值对互换，即把<单词，出现次数> 改为 <出现次数，单词>，
		JavaPairRDD<Integer, String> temp = results.mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {
			private static final long serialVersionUID = -7749760293871765524L;

			public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {
				return new Tuple2<Integer, String>(tuple._2, tuple._1);
			}
		});
		
		//根据出现次数排序，排完序后，再次互换键值对，即最终还是以单词做为key
		JavaPairRDD<String, Integer> sorted = temp.sortByKey(false)
				.mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() {
			private static final long serialVersionUID = -1L;

			public Tuple2<String, Integer> call(Tuple2<Integer,String> tuple) throws Exception {
				return new Tuple2<String, Integer>(tuple._2, tuple._1);
			}
		});
		
		sorted.collect().forEach((tuple) -> {
			System.out.println(tuple._1 + "," + tuple._2);
		});
		/*sorted.collect().foreach(new VoidFunction<Tuple2<String,Integer>>() {
			
			private static final long serialVersionUID = 9073409118436911837L;

			public void call(Tuple2<String, Integer> tuple) throws Exception {
				System.out.println("word: " + tuple._1 + " Count: " + tuple._2);
			}
		});*/
		
		ctx.close();
	}
}
